package com.iflytek.iot.hadoop.dfs.util;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

/**
 * hdfs文件操作的工具类
 *
 * @author yu 2019/11/28.
 */
public class HdfsUtil {

    private static final Logger LOGGER = LoggerFactory.getLogger(HdfsUtil.class);

    /**
     * @param conf     配置
     * @param hdfsUri  远程http url
     * @param userName 使用hadoop的用户
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    public static FileSystem getFileSystem(Configuration conf, String hdfsUri, String userName) {
        FileSystem fs = null;
        if (hdfsUri == null) {
            try {
                fs = FileSystem.get(conf);
            } catch (IOException e) {
                LOGGER.error("Failed to perform hdfs operation", e);
            }
        } else {
            //默认采用root用户
            userName = StringUtils.isBlank(userName) ? "root" : userName;
            try {
                fs = FileSystem.get(URI.create(hdfsUri), conf, userName);
            } catch (IOException | InterruptedException e) {
                LOGGER.error("Failed to perform hdfs operation", e);
            }
        }
        return fs;
    }

    /**
     * 列出hdfs指定路径上的文件
     *
     * @param path       路径
     * @param fileSystem hdfs文件系统
     * @return list of file name
     * @throws IOException
     */
    public static List<String> getFileList(String path, FileSystem fileSystem) throws IOException {
        FileStatus[] fsStatus = fileSystem.listStatus(new Path(path));
        List<String> files = new ArrayList<>(fsStatus.length);
        for (FileStatus fileStatus : fsStatus) {
            files.add(fileStatus.getPath().toString());
        }
        return files;
    }


    /**
     * 在hdfs上创建文件
     *
     * @param filePath
     * @param contents
     * @param fileSystem
     * @throws IOException
     */
    public static String createFile(String filePath, String contents, FileSystem fileSystem) throws IOException {
        Path path = new Path(filePath);
        if (fileSystem.exists(path)) {
            return null;
        }
        FSDataOutputStream outputStream = fileSystem.create(path);
        outputStream.writeUTF(contents);
        fileSystem.close();
        outputStream.close();
        return filePath;
    }

    /**
     * 创建一个可追加的文件
     *
     * @param filePath
     * @param contents
     * @param fileSystem hdfs文件系统
     * @throws IOException
     */
    public static void createFileAppendable(String filePath, String contents, FileSystem fileSystem) throws IOException {
        Path path = new Path(filePath);
        FSDataOutputStream fileOutputStream = null;
        try {
            if (fileSystem.exists(path)) {
                fileOutputStream = fileSystem.append(path);
                fileOutputStream.writeUTF(contents);
                fileOutputStream.writeUTF("\n");
            } else {
                fileOutputStream = fileSystem.create(path);
                fileOutputStream.writeUTF(contents);
                fileOutputStream.writeUTF("\n");
            }
        } finally {
            if (fileSystem != null) {
                fileSystem.close();
            }
            if (fileOutputStream != null) {
                fileOutputStream.close();
            }
        }
    }

    /**
     * 获取文件内容
     *
     * @param remoteFile hdfs上的文件
     * @param fileSystem hdfs文件系统
     * @return string file content
     * @throws IOException
     */
    public static String download(String remoteFile, FileSystem fileSystem) throws IOException {
        Path path = new Path(remoteFile);
        String str;
        FSDataInputStream fsdis = null;
        OutputStream baos = new ByteArrayOutputStream();
        try {
            fsdis = fileSystem.open(path);
            IOUtils.copyBytes(fsdis, baos, 4096, false);
            str = baos.toString();
        } finally {
            IOUtils.closeStream(fsdis);
            fileSystem.close();
        }
        return str;
    }

    /**
     * 删除文件
     *
     * @param fileName   文件名
     * @param fileSystem hdfs文件系统
     * @return boolean
     */
    public static boolean delete(String fileName, FileSystem fileSystem) {
        boolean found = false;
        try {
            Path fspath = new Path(fileName);
            found = fileSystem.exists(fspath);
            if (!found) {
                LOGGER.error("Deletion is not carried out as file {} does not exist in hdfs", fileName);
                return false;
            } else {
                LOGGER.debug("Deleting file {} in hdfs ", fileName);
                fileSystem.delete(fspath, true);
            }
        } catch (IOException e) {
            LOGGER.error("Failed to perform hdfs operation", e);
        }
        return found;
    }

    /**
     * 在hdfs上创建目录
     *
     * @param dir        目录地址
     * @param fileSystem hdfs文件系统
     * @return
     */
    public static boolean createDir(String dir, FileSystem fileSystem) {
        try {
            if (!fileSystem.exists(new Path(dir))) {
                fileSystem.mkdirs(new Path(dir));
                fileSystem.close();
                return true;
            }
        } catch (IOException e) {
            LOGGER.error("Failed to create dir", e);
        }
        return false;
    }
}
